package com.young.oneway;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;

import java.nio.charset.StandardCharsets;
import java.util.List;

/**
 * 消费者
 */
public class OneWayConsumer {
    public static void main(String[] args) throws MQClientException {
        // 1、创建消费者实例，并传递消费者组名称（推送模型的消费者组）
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("oneway_consumer_group");
        // 2、消费者设组设置nameServer地址
        consumer.setNamesrvAddr("192.168.188.101:9876;192.168.188.102:9876;");
        // 3、订阅topic
        consumer.subscribe("oneway_topic","*");
        // 4、注册消息监听者，监听并处理消息
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                list.forEach(message -> {
                    String messageBody = new String(message.getBody(),StandardCharsets.UTF_8);
                    System.out.println("接收到消息, topic:"+message.getTopic()+"Tags："+message.getTags()+"消息内容："+messageBody);
                });
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        // 5、启动消费者
        consumer.start();

        System.out.println("Consumer Started");

    }
}
